package com.voice.core.channel.tcp.impl;

import com.voice.core.channel.IOContext;
import com.voice.core.channel.Receiver;
import com.voice.core.channel.Sender;
import com.voice.core.connection.Connection;
import com.voice.core.connection.frame.Frame;
import com.voice.core.connection.frame.FrameFactory;
import com.voice.core.connection.frame.StreamFrame;
import com.voice.core.utils.CloseUtil;

import java.io.Closeable;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.Arrays;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

public class SocketChannelAdapter implements Sender, Receiver, Closeable {

    private AtomicBoolean isClosed = new AtomicBoolean(false);
    private AtomicBoolean isSend = new AtomicBoolean(false);
    private IOContext ioContext;
    private SocketChannel socketChannel;
    private Connection connection;
    private BlockingQueue<Frame> frames = new ArrayBlockingQueue<Frame>(100);
    private Frame frame;
    private ByteBuffer receiveBuffer = ByteBuffer.allocate(512);
    private ByteBuffer sendBuffer = ByteBuffer.allocate(512);


    public SocketChannelAdapter(IOContext ioContext, SocketChannel socketChannel, Connection connection) {
        this.ioContext = ioContext;
        this.socketChannel = socketChannel;
        this.connection = connection;
        try {
            socketChannel.configureBlocking(false);
        } catch (IOException e) {
            this.close();
        }
        receiveBuffer.limit(0);
        receiverFrameAsync();
        sendAsync();
    }

    public void setConnection(Connection connection) {
        this.connection = connection;
    }

    @Override
    public void receiverFrameAsync() {
        ioContext.registerInput(socketChannel, () -> {
            if (isClosed.get()) return;
            try {
                if (receiveBuffer.hasRemaining()) {
                    receiveBuffer.compact();
                } else receiveBuffer.clear();
                socketChannel.read(receiveBuffer);
                receiveBuffer.flip();
                while (receiveBuffer.hasRemaining()) {
                    if (frame == null) {
                        frame = FrameFactory.newFrame(receiveBuffer);
                    }
                    if (frame == null) {
                        return;
                    };
                    frame.read(receiveBuffer);
                    if (frame.readFinish()) {
                        frame.onFinish(connection);
                        if (!(frame instanceof StreamFrame)) {
                            frame = null;
                        }
                    }
                }
            } catch (Exception e) {
                this.close();
            }
        });
    }

    @Override
    public void close() {
        if (isClosed.compareAndSet(false, true)) {
            ioContext.unRegisterInput(socketChannel);
            ioContext.unRegisterOutput(socketChannel);
            connection.onConnectionClosed();
            CloseUtil.close(socketChannel);
        }
    }

    @Override
    public void writeFrameAsync(Frame frame) {
        try {
            frames.put(frame);
            if(isSend.compareAndSet(false,true)){
                sendAsync();
            }
        } catch (InterruptedException e) {
            writeFrameAsync(frame);
        }
    }

    private void sendAsync() {
        ioContext.registerOutput(socketChannel, () -> {
            if (isClosed.get()) return;
            Frame frame = frames.poll();
            while (frame != null) {
                try {
                    sendBuffer.clear();
                    frame.write(sendBuffer);
                    sendBuffer.flip();
                    while (sendBuffer.hasRemaining()) {
                        socketChannel.write(sendBuffer);
                    }
                } catch (Exception e) {
                    this.close();
                    break;
                }
                if (frame.writeFinish()) {
                    frame = frames.poll();
                }
            }
            isSend.set(false);
        });
    }

    public SocketChannel getSocketChannel() {
        return socketChannel;
    }

}
